#pragma once
#include "mq_channal.hpp"
#include "muduo/protoc/codec.h"
#include "muduo/protoc/dispatcher.h"
#include "muduo/net/TcpServer.h"
#include "muduo/net/EventLoop.h"

#include "muduo/net/EventLoop.h"
#include "muduo/net/TcpServer.h"

#include "mq_connection.hpp"
#include "mq_virtualhost.hpp"
#include "mq_consumer.hpp"
#include "../mqcommon/mq_proto.pb.h"
#include "../mqcommon/mq_msg.pb.h"
#include <memory>
namespace mq_zxws
{
    class Server{
    public:
        #define DBFILE "./meta.db"
        using MessagePtr = std::shared_ptr<google::protobuf::Message>;
        Server(int port, const std::string& basedir)
        :_server(&_baseloop, muduo::net::InetAddress("0.0.0.0", port), "server", muduo::net::TcpServer::kReusePort)
        ,_dispatcher(std::bind(&Server::onUnknownMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3))
        ,_codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)))
        ,_virtual_host(std::make_shared<VirtualHost>("host", basedir, basedir + DBFILE))
        ,_connection_manager(std::make_shared<ConnectionManager>())
        ,_consumer_manager(std::make_shared<ConsumerManager>())
        ,_threadpool(std::make_shared<ThreadPool>())
        {
            auto allqueues = _virtual_host->AllQueues();
            for (auto it : allqueues) {
                _consumer_manager->initComsumerQueue(it.first);
            }
            _dispatcher.registerMessageCallback<openChannalRequest>(std::bind(&Server::onOpenChannal, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<closeChannalRequest>(std::bind(&Server::onCloseChannal, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<declareExchangeRequest>(std::bind(&Server::onDeclareExchange, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<deleteExchangeRequest>(std::bind(&Server::onDeleteExchange, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<declareQueueRequest>(std::bind(&Server::onDeclareQueue, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<deleteQueueRequest>(std::bind(&Server::onDeleteQueue, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<basicConsumerRequest>(std::bind(&Server::onBasicConsumer, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<basicCancelRequest>(std::bind(&Server::onBasicCancel, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<basicPublishRequest>(std::bind(&Server::onBasicPublish, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<basicAckRequest>(std::bind(&Server::onBasicAck, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<queueBindRequest>(std::bind(&Server::onQueueBind, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<queueUnBindRequest>(std::bind(&Server::onQueueUnBind, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _server.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec.get(), std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _server.setConnectionCallback(std::bind(&Server::onConnetion, this, std::placeholders::_1));
        }
        void start(){
            _server.start();
            _baseloop.loop();
        }
    private:
        // 打开信道
        void onOpenChannal(const muduo::net::TcpConnectionPtr& conn, const openChannalRequestPtr& req, muduo::Timestamp){
            DLOG("接收到打开信道的请求");
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if(mconn.get() == nullptr){
                DLOG("该信道对应的连接不存在");
                conn->shutdown();
                return;
            }
            return mconn->openChannal(req);
        }
        // 关闭信道
        void onCloseChannal(const muduo::net::TcpConnectionPtr& conn, const closeChannalRequestPtr& req, muduo::Timestamp){
            DLOG("接收到关闭信道的请求");
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if(mconn.get() == nullptr){
                DLOG("关闭信道时，没有找到连接对应的Connection对象");
                conn->shutdown();
                return;
            }
            return mconn->closeChannal(req);
        }
        // 声明交换机
        void onDeclareExchange(const muduo::net::TcpConnectionPtr& conn, const declareExchangeRequestPtr& req, muduo::Timestamp){
            DLOG("接收到声明交换机的请求");
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if(mconn.get() == nullptr){
                DLOG("声明交换机时，没有找到连接对应的Connection对象");
                conn->shutdown();
                return;
            }
            Channal::ptr cp = mconn->getChannal(req->cid());
            if(cp.get() == nullptr){
                DLOG("声明交换机时，没有找到连接对应的Connection对象");
                return;
            }
            return cp->declareExchange(req);
        }
        // 删除交换机
        void onDeleteExchange(const muduo::net::TcpConnectionPtr& conn, const deleteExchangeRequestPtr& req, muduo::Timestamp){
            DLOG("接收到删除交换机的请求");
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if(mconn.get() == nullptr){
                DLOG("删除交换机时，没有找到连接对应的Connection对象");
                conn->shutdown();
                return;
            }
            Channal::ptr cp = mconn->getChannal(req->cid());
            if(cp.get() == nullptr){
                DLOG("删除交换机时，没有找到连接对应的Connection对象");
                return;
            }
            return cp->deleteExchange(req);
        }
        // 声明队列
        void onDeclareQueue(const muduo::net::TcpConnectionPtr& conn, const declareQueueRequestPtr& req, muduo::Timestamp){
            DLOG("接收到声明队列的请求");
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if(mconn.get() == nullptr){
                DLOG("声明队列时，没有找到连接对应的Connection对象");
                conn->shutdown();
                return;
            }
            Channal::ptr cp = mconn->getChannal(req->cid());
            if(cp.get() == nullptr){
                DLOG("声明队列时，没有找到连接对应的Connection对象");
                return;
            }
            return cp->declareQueue(req);
        }
        // 删除队列
        void onDeleteQueue(const muduo::net::TcpConnectionPtr& conn, const deleteQueueRequestPtr& req, muduo::Timestamp){
            DLOG("接收到删除队列的请求");
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if(mconn.get() == nullptr){
                DLOG("删除队列时，没有找到连接对应的Connection对象");
                conn->shutdown();
                return;
            }
            Channal::ptr cp = mconn->getChannal(req->cid());
            if(cp.get() == nullptr){
                DLOG("删除队列时，没有找到连接对应的Connection对象");
                return;
            }
            return cp->deleteQueue(req);
        }
        // 订阅消息
        void onBasicConsumer(const muduo::net::TcpConnectionPtr& conn, const basicConsumerRequestPtr& req, muduo::Timestamp){
            DLOG("接收到订阅消息的请求");
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if(mconn.get() == nullptr){
                DLOG("订阅信息时，没有找到连接对应的Connection对象");
                conn->shutdown();
                return;
            }
            Channal::ptr cp = mconn->getChannal(req->cid());
            if(cp.get() == nullptr){
                DLOG("订阅信息时，没有找到连接对应的Connection对象");
                return;
            }
            return cp->basicConsumer(req);
        }
        // 取消订阅
        void onBasicCancel(const muduo::net::TcpConnectionPtr& conn, const basicCancelRequestPtr& req, muduo::Timestamp){
            DLOG("接收到取消订阅的请求");
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if(mconn.get() == nullptr){
                DLOG("取消订阅信息时，没有找到连接对应的Connection对象");
                conn->shutdown();
                return;
            }
            Channal::ptr cp = mconn->getChannal(req->cid());
            if(cp.get() == nullptr){
                DLOG("取消订阅信息时，没有找到连接对应的Connection对象");
                return;
            }
            return cp->basicCancel(req);
        }
        // 发布消息
        void onBasicPublish(const muduo::net::TcpConnectionPtr& conn, const basicPublishRequestPtr& req, muduo::Timestamp){
            DLOG("接收到发布消息的请求");
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if(mconn.get() == nullptr){
                DLOG("发布消息时，没有找到连接对应的Connection对象");
                conn->shutdown();
                return;
            }
            Channal::ptr cp = mconn->getChannal(req->cid());
            if(cp.get() == nullptr){
                DLOG("发布消息时，没有找到连接对应的Connection对象");
                return;
            }
            return cp->basicPublish(req);
        }
        // 确认消息
        void onBasicAck(const muduo::net::TcpConnectionPtr& conn, const basicAckRequestPtr& req, muduo::Timestamp){
            DLOG("接收到确认消息的请求");
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if(mconn.get() == nullptr){
                DLOG("确认信息时，没有找到连接对应的Connection对象");
                conn->shutdown();
                return;
            }
            Channal::ptr cp = mconn->getChannal(req->cid());
            if(cp.get() == nullptr){
                DLOG("确认信息时，没有找到连接对应的Connection对象");
                return;
            }
            return cp->basicAck(req);
        }
        // 绑定
        void onQueueBind(const muduo::net::TcpConnectionPtr& conn, const queueBindRequestPtr& req, muduo::Timestamp){
            DLOG("接收到绑定的请求");
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if(mconn.get() == nullptr){
                DLOG("绑定队列时，没有找到连接对应的Connection对象");
                conn->shutdown();
                return;
            }
            Channal::ptr cp = mconn->getChannal(req->cid());
            if(cp.get() == nullptr){
                DLOG("绑定队列时，没有找到连接对应的Connection对象");
                return;
            }
            return cp->queueBind(req);
        }
        // 解除绑定
        void onQueueUnBind(const muduo::net::TcpConnectionPtr& conn, const queueUnBindRequestPtr& req, muduo::Timestamp){
            DLOG("接收到解除绑定的请求");
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if(mconn.get() == nullptr){
                DLOG("解除绑定时，没有找到连接对应的Connection对象");
                conn->shutdown();
                return;
            }
            Channal::ptr cp = mconn->getChannal(req->cid());
            if(cp.get() == nullptr){
                DLOG("解除绑定时，没有找到连接对应的Connection对象");
                return;
            }
            return cp->queueUnBind(req);
        }
        void onConnetion(const muduo::net::TcpConnectionPtr & conn){
            if(conn->connected()){
                _connection_manager->newConnection(conn, _codec, _consumer_manager, _virtual_host, _threadpool);
            }else{
                _connection_manager->deleteConnection(conn);
            }
        }
        void onUnknownMessage(const muduo::net::TcpConnectionPtr& conn, const MessagePtr& message, muduo::Timestamp){
            DLOG("onUnknownMessage: %s", message->GetTypeName().c_str());
            conn->shutdown();
        }
    private:
        muduo::net::EventLoop _baseloop;
        muduo::net::TcpServer _server;
        ProtobufDispatcher _dispatcher;
        ProtobufCodecPtr _codec;
        VirtualHost::ptr _virtual_host;
        ConnectionManager::ptr _connection_manager;
        ConsumerManager::ptr _consumer_manager;
        ThreadPool::ptr _threadpool;
    };
}